#pragma once
#include "muduo/proto/codec.h"
#include "muduo/net/TcpConnection.h"

#include "../mqcommon/helper.hpp"
#include "../mqcommon/logger.hpp"
#include "../mqcommon/msg.pb.h"
#include "../mqcommon/request.pb.h"
#include "../mqcommon/threadpool.hpp"
#include "router.hpp"
#include "consumer.hpp"
#include "virtualhost.hpp"
#include <mutex>

using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
using openChannelRequestPtr = std::shared_ptr<openChannelRequest>;
using closeChannelRequestPtr = std::shared_ptr<closeChannelRequest>;
using declareExchangeRequestPtr = std::shared_ptr<declareExchangeRequest>;
using deleteExchangeRequestPtr = std::shared_ptr<deleteExchangeRequest>;
using declareMsgQueueRequestPtr = std::shared_ptr<declareMsgQueueRequest>;
using deleteMsgQueueRequestPtr = std::shared_ptr<deleteMsgQueueRequest>;
using bindRequestPtr = std::shared_ptr<bindRequest>;
using unBindRequestPtr = std::shared_ptr<unBindRequest>;
using basicPublishRequestPtr = std::shared_ptr<basicPublishRequest>;
using basicAckRequestPtr = std::shared_ptr<basicAckRequest>;
using basicConsumerRequestPtr = std::shared_ptr<basicConsumerRequest>;
using basicCancelRequestPtr = std::shared_ptr<basicCancelRequest>;
using pullOnceRequestPtr = std::shared_ptr<pullOnceRequest>;
class channel
{
public:
    using Ptr = std::shared_ptr<channel>;
    channel(const std::string& id, const muduo::net::TcpConnectionPtr& conn, const ProtobufCodecPtr& codec
            , const QueueConsumerManager::Ptr& qcManager, const VirtualHost::Ptr& host, const ThreadPool::Ptr& threadpool)
        :_cid(id), _conn(conn), _codec(codec), _qcManager(qcManager), _host(host), _threadpool(threadpool)
    {}
    ~channel()
    {
        if(_consumer.get())
            _qcManager->remove(_consumer->queueName, _consumer->tag);
    }
    void declareExchange(const declareExchangeRequestPtr& req)
    {
        _host->declareExchange(req->exchange_name(), req->exchange_type(), req->durable(), req->auto_delete(), req->args());
        basicResponse(req->cid(), req->rid(), true);
    }
    void deleteExchange(const deleteExchangeRequestPtr& req)
    {
        _host->deleteExchange(req->exchange_name());
        basicResponse(req->cid(), req->rid(), true);
    }

    void declareMsgQueue(const declareMsgQueueRequestPtr& req)
    {
        _host->declareMsgQueue(req->msgqueue_name(), req->durable(), req->exclusive(), req->auto_delete(), req->args()); //队列初始化消息，_host里做了
        _qcManager->initQueueConsumer(req->msgqueue_name());
        basicResponse(req->cid(), req->rid(), true);
    }
    void deleteMsgQueue(const deleteMsgQueueRequestPtr& req)
    {
        _host->deleteMsgQueue(req->msgqueue_name());
        _qcManager->destoryQueueConsumer(req->msgqueue_name());
        basicResponse(req->cid(), req->rid(), true);
    }

    void bind(const bindRequestPtr& req)        
    {
        bool ok = _host->bind(req->exchange_name(), req->msgqueue_name(), req->binding_key());  //只有它会失败
        basicResponse(req->cid(), req->rid(), ok);
    }
    void unBind(const unBindRequestPtr& req)
    {
        _host->unBind(req->exchange_name(), req->msgqueue_name());
        basicResponse(req->cid(), req->rid(), true);
    }

    void basicPublish(const basicPublishRequestPtr& req) //生产者发布消息，服务器推送消息给消费者
    {
        Exchange::Ptr exchangePtr = _host->selectExchange(req->exchange_name());
        if(exchangePtr.get() == nullptr)        //交换机都没有，无法发布
            return basicResponse(req->cid(), req->rid(), false);
        
        BasicProperty* basicPropertyPtr = nullptr;
        std::string routing_key;
        if(req->has_bp())
        {
            basicPropertyPtr = req->mutable_bp();
            routing_key = basicPropertyPtr->routing_key();
        }

        MsgQueueBindingMap msgQueueBindingMap = _host->exchangeBindings(req->exchange_name());
        for(auto& msgQueueBinding : msgQueueBindingMap)                 //历史就是无法推送  //这里是看能不能找到对应的队列名字
        {
            if(Router::route(exchangePtr->type, routing_key, msgQueueBinding.second->bindingKey)) //req的routing_key与每一个队列的binding_key匹配
            {
                _host->basicPublish(msgQueueBinding.first, basicPropertyPtr, req->body());          //消息的发布  
                // _threadpool->Push(&channel::consume, this, msgQueueBinding.first); //这样绑定不进去
                // _threadpool->Push(std::bind(&channel::consume, this, msgQueueBinding.first));       
                while(true)
                {
                    auto task_t = std::bind(&channel::consume, this, msgQueueBinding.first);            //消息的推送,通知消费
                    std::future<bool> fu = _threadpool->Push(task_t);
                    if(fu.get() == false) break;    
                }
            }
        }
        basicResponse(req->cid(), req->rid(), true);
    }
    void basicAck(const basicAckRequestPtr& req)
    {
        _host->basicAck(req->msgqueue_name(), req->msgid());
        return basicResponse(req->cid(), req->rid(), true);
    }

    void basicConsumer(const basicConsumerRequestPtr& req)
    {
        bool ok = _host->existsMsgQueue(req->msgqueue_name());
        if(ok == false)
        {
            DLOG("队列：%s,不存在", req->msgqueue_name().c_str());
            return basicResponse(req->cid(), req->rid(), false);
        }
        auto cb = std::bind(&channel::callback, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3);
        _consumer = _qcManager->create(req->consumer_tag(), req->msgqueue_name(), req->auto_ack(), cb);        /////////////////////////////
        DLOG("%s", req->cid().c_str()); 
        basicResponse(req->cid(), req->rid(), true);       
    }
    void basicCancel(const basicCancelRequestPtr& req)
    {
        _qcManager->remove(req->consumer_tag(), req->msgqueue_name());
        basicResponse(req->cid(), req->rid(), true);
    }
    void pullOnce(const pullOnceRequestPtr& req)
    {
        while(true)
        {
            auto task_t = std::bind(&channel::consume, this, req->msgqueue_name());            //消息的推送,通知消费
            std::future<bool> fu = _threadpool->Push(task_t);
            if(fu.get() == false) break;    
        }
        basicResponse(req->cid(), req->rid(), true);
    }
private:
    void callback(const std::string& tag, const BasicProperty* bp, const std::string& body)
    {
        basicConsumerResponse resp;
        resp.set_cid(_cid);
        resp.set_consumer_tag(tag);
        DLOG("服务端走到这");
        resp.mutable_bp()->set_id(bp->id());                //必须到每一根，resp他自己要有一份
        resp.mutable_bp()->set_delivery_mode(bp->delivery_mode());
        resp.mutable_bp()->set_routing_key(bp->id());
        resp.set_body(body);
        DLOG("发出的内容%s",body.c_str());
        _codec->send(_conn, resp);
        if(_consumer->autoAck)      //是否需要自动删除
            _host->basicAck(_consumer->queueName, bp->id());                                                        //自动删除功能
    }
    bool consume(const std::string& qname)
    {
        Consumer::Ptr consumerPtr = _qcManager->choose(qname);
        if(consumerPtr.get() == nullptr)
        {
            DLOG("无法推送，这个队列没有消费者");
            return false;
        }
        wzf::MessagePtr messagePtr = _host->basicConsume(qname);         //只有这个智能指针是放在类外定义的
        if(messagePtr.get() == nullptr)
        {
            DLOG("无法推送，队列没有消息");
            return false;
        }
        DLOG("准备发回给消费者端")
        consumerPtr->callback(consumerPtr->tag, messagePtr->mutable_paylode()->mutable_property(), messagePtr->paylode().body());
        DLOG("准备删除内存中的备份")
        
        if(consumerPtr->autoAck == true)
            _host->basicAck(qname, messagePtr->paylode().property().id());
        return true;
    }
    void basicResponse(const std::string& cid, const std::string& rid, bool ok)
    {
        basicCommonResponse resp;
        resp.set_cid(cid);
        resp.set_rid(rid);
        resp.set_ok(ok);
        _codec->send(_conn, resp);
    }
private:
    std::string _cid;                            //信道的标识
    Consumer::Ptr _consumer;                     //信道对应的消费者
    muduo::net::TcpConnectionPtr _conn;          //信道对应的链接    
    ProtobufCodecPtr _codec;                     //解析protobuf协议
    QueueConsumerManager::Ptr _qcManager;        //消费者管理
    VirtualHost::Ptr _host;                      //交换机，队列，绑定的操作
    ThreadPool::Ptr _threadpool;
};

class channelManager
{
public:
    using Ptr = std::shared_ptr<channelManager>;
    channelManager() {}
    ~channelManager() {}

    bool openChannel(const std::string& id, const muduo::net::TcpConnectionPtr& conn, const ProtobufCodecPtr& codec
            , const QueueConsumerManager::Ptr& qcManager, const VirtualHost::Ptr& host, const ThreadPool::Ptr& threadpool)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        auto it = _channels.find(id);
        if(it != _channels.end())
        {
            DLOG("%s信道已经存在", id.c_str());
            return false;
        }
        channel::Ptr channelPtr = std::make_shared<channel>(id, conn, codec, qcManager, host, threadpool);
        _channels.insert({id, channelPtr});
        return true;
    }

    bool closeChannel(const std::string id)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        _channels.erase(id);
        return true;
    }

    channel::Ptr getChannel(const std::string& id)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        auto it = _channels.find(id);
        if(it ==  _channels.end())
        {
            DLOG("%s信道不存在", id.c_str());
            return channel::Ptr();
        }
        return it->second;
    }
private:
    std::mutex _mutex;
    std::unordered_map<std::string, channel::Ptr> _channels;
};